/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZLoop;
import org.zeromq.ZMQ; 
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZMsg;

public class CSBQueue extends CSBServerBase implements Runnable {
	private String topic;
	private String groupId;
	private String zkConn;
	private String frontConn;
	private String backConn;
	private int numThread;
	private int bufferSize;
	private int timeout;
    private int retry;
	private Socket frontend;
	private Socket backend;        
	private List<Worker> workers;
    private final String id = CSBUtils.sockId();
    private int mode;
    private ErrorHandler errHandler;
    
    private static Logger LOG = Logger.getLogger(CSBQueue.class.getName());
	
    private static class Worker {
        ZFrame address;          
        String identity;         
        long expiry;            
        
        protected Worker(ZFrame address) {
            this.address = address;
            identity = new String(address.getData());
            expiry = System.currentTimeMillis() + CSB.HEARTBEAT_INTERVAL*CSB.HEARTBEAT_LIVENESS;
        }

        protected void ready(List<Worker> workers) {
            Iterator<Worker> it = workers.iterator();
            while (it.hasNext()) {
                Worker worker = it.next();
                if (identity.equals(worker.identity)) {
                    it.remove();
                    break;
                }
            }
            workers.add(this);
        }
        
        protected static ZFrame next(List<Worker> workers) {
            Worker worker = workers.remove(0);
            ZFrame frame = worker.address;
            return frame;
        }

        protected static ZFrame get(List<Worker> workers, String identity) {
        	ZFrame frame = null;
        	Iterator<Worker> it = workers.iterator();
            while (it.hasNext()) {
                Worker worker = it.next();
                if (identity.equals(worker.identity)) {
                	frame = worker.address;
                    it.remove(); 
                    break;
                }
            }            
            return frame;
        }

        protected static void purge(List<Worker> workers) {
            Iterator<Worker> it = workers.iterator();
            while (it.hasNext()) {
                Worker worker = it.next();
                if (System.currentTimeMillis() < worker.expiry) {
                    break;
                }
                it.remove();
            }
        }
    };
		
    public CSBQueue(String topic, String groupId, String zkConn, 
    	String frontConn, String backConn, int numThread, 
    		int bufferSize, int timeout, int retry, int mode, 
    		ErrorHandler errHandler, int perfInterval) {
    	super(perfInterval);
    	this.topic = topic;
        this.groupId = groupId;
        this.zkConn = zkConn;
    	this.frontConn = frontConn;
    	this.backConn = backConn;
    	this.numThread= numThread;
    	this.bufferSize = bufferSize;
    	this.timeout = timeout;
        this.retry = retry;
        this.mode = mode;
        this.errHandler = errHandler;
        LOG.info("queue created: " + id);
    }

    private IZLoopHandler front = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
            ZMsg msg = ZMsg.recvMsg(frontend);
            if (msg != null) {
            	ZFrame address = msg.unwrap();
            	String s = new String(address.getData());
            	if (msg.size() == 2) {
            		ZFrame id = msg.pop();
            		ZFrame dest = Worker.get(workers, new String(id.getData()));                	
            		if (dest == null) {
            			ZMsg ack = ZMsg.newStringMsg(CSB.ROUTER_UNREACHABLE);
            			ack.wrap(address);
            			ack.send(frontend);
            			msg.destroy();
            		} else {
            			msg.wrap(address);
            			msg.push(dest);
            			msg.send(backend);
            		}
            	} else {     
            		msg.wrap(address);
            		msg.push(Worker.next(workers));
            		msg.send(backend);
            	}
            
            	mark(s);
            	if (workers.size() == 0) {
            		loop.removePoller(new PollItem(frontend, 0));
            	}
            	
            	Worker.purge(workers);
        	} else {
        		LOG.error(id + ": frontend receiving interrupted.");
        	}
            return 0;
        }
    };
    
    private IZLoopHandler back = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
            ZMsg msg = ZMsg.recvMsg(backend);
            if (msg != null) {
            	ZFrame address = msg.unwrap();
            	Worker worker = new Worker(address);
            	worker.ready(workers);
            	
            	if (workers.size() == 1) {
            		PollItem newItem = new PollItem(frontend, ZMQ.Poller.POLLIN);
            		loop.addPoller(newItem, front, null);
            	}
            
            	if (msg.size() == 1) {
            		ZFrame frame = msg.pop();
            		String data = new String(frame.getData());
            		if (!data.equals(CSB.WORKER_READY)
            				&&  !data.equals(CSB.WORKER_HEARTBEAT)) {
            			LOG.warn(id + ": malformed message from worker - " + data);
            		}
            		msg.destroy();
            	}
            	else {
            		msg.send(frontend);
            	}
            	
            	Worker.purge(workers);
            } else {
            	LOG.error(id + ": backend receiving interrupted.");
            }
        	return 0;
        }
    };
    
    private IZLoopHandler heart = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
        	for (Worker worker: workers) {                    
                worker.address.send(backend, ZFrame.REUSE + ZFrame.MORE);
                ZFrame frame = new ZFrame(CSB.WORKER_HEARTBEAT);
                frame.send(backend, 0);
            }
        	return 0;
        }
    };
    
    public void run() {
    	perf();

        workers = new ArrayList<Worker>();        
    	ZContext ctx = new ZContext();
        frontend = ctx.createSocket(ZMQ.ROUTER);
        backend = ctx.createSocket(ZMQ.ROUTER);        
        frontend.bind(frontConn);          
        backend.bind(backConn);        
        LOG.info(id + ": binding backend via " + backConn + "...");
        LOG.info(id + ": binding frontkend via " + frontConn + "...");

        Receiver receiver = new Receiver
        	(topic, groupId, zkConn, numThread, bufferSize, 
        		ctx, frontConn, timeout, retry, mode, errHandler);        
        new Thread(receiver).start();
        LOG.info(id + ": receiver starting...");
        	
        ZLoop loop = new ZLoop();
    	PollItem poller = new PollItem(backend, Poller.POLLIN);
    	loop.addPoller(poller, back, null);
    	loop.addTimer(CSB.HEARTBEAT_INTERVAL, 0, heart, null);        		
    	loop.start();
    	loop.destroy();
    	LOG.error(id + ": loop interrupted, exiting...");
    	
        while (workers.size() > 0) {
        	workers.remove(0);
        }        
        workers.clear();
        ctx.destroy();        
    }
}

